package consumer

import (
	"errors"
	"gamesdk/common/log"
	"sync"
	"time"

	"github.com/Shopify/sarama"
)

var (
	g_consumer sarama.Consumer = nil
	wg         sync.WaitGroup
	g_ip       []string
)

type recvMsg func(partition int32, offset int64, key []byte, value []byte)

func InitConsumer(ips []string) error {
	config := sarama.NewConfig()
	config.Net.ReadTimeout = 10 * time.Second
	config.Consumer.Return.Errors = true
	//config.Version = sarama.V0_11_0_2

	consumer, err := sarama.NewConsumer(ips, config)
	if err != nil {
		log.Error("Failed to start consumer:" + err.Error())
		return err
	}
	g_consumer = consumer
	g_ip = ips
	log.Info("connect consumer kafka success ips:%v", ips)
	return err
}

func RecvMsg(topic string, offset int64, f recvMsg) error {
	partitionList, err := g_consumer.Partitions(topic)
	if err != nil {
		log.Error("Failed to get the list of partitions: %v", err)
		return err
	}
	if len(partitionList) <= 0 {
		log.Error("Failed to get the list of partitions: %v", err)
		return err
	}
	c_offset := offset
	var pc sarama.PartitionConsumer
	ok := false
	for _, p := range partitionList {
		pc, err = g_consumer.ConsumePartition(topic, p, offset)
		if err != nil {
			log.Error("Failed to start consumer for partition %d: %v", p, err)
			continue
		}
		ok = true
		break
	}
	if ok == false {
		return errors.New("Failed to start consumer for partition error")
	}
	for {
		select {
		case msg := <-pc.Messages():
			f(msg.Partition, msg.Offset, msg.Key, msg.Value)
			c_offset = msg.Offset + 1
		case err := <-pc.Errors():
			log.Error("err :%s\n", err.Error())
			goto A
		}
	}
A:

	for {
		pc.Close()
		time.Sleep(time.Second * 2)
		config := sarama.NewConfig()
		config.Consumer.Return.Errors = true
		consumer, err := sarama.NewConsumer(g_ip, config)
		if err != nil {
			log.Error("Failed to start consumer:" + err.Error())
			continue
		}
		g_consumer = consumer
		log.Info("重连成功")
		break
	}
	return RecvMsg(topic, c_offset, f)
}

func Close() {
	if g_consumer == nil {
		g_consumer.Close()
		g_consumer = nil
	}
}
